package com.cloudsubassemblykafka10024.topic;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.internals.Topic;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description: Kafka Topic 管理
 * @Author zx
 * @Date 2024/12/24 13:49
 * @Version 1.0
 */
public class KafkaTopicDMI {

    public static void createTopic() throws ExecutionException, InterruptedException {
        // 创建 Kafka Admin Client
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.18.220:9092");

        KafkaAdminClient adminClient = (KafkaAdminClient) KafkaAdminClient.create(properties);

        // 创建 Topic
//         newCreateTopic(adminClient, "messageDML003");

        // 删除 Topic
         //adminClient.deleteTopics(Arrays.asList("topic"));

        // 查看Topic列表
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        listTopicsResult.names().get().forEach(System.out::println);

        // 查看Topic详情
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(
                Arrays.asList("messageDML001"));
        describeTopicsResult.all().get().forEach((k, v) -> System.out.println(k + ":" + v));


        // 关闭 Admin Client
        adminClient.close();

    }

    /**
     * 创建 Topic
     * @param adminClient
     */
    private static void newCreateTopic(KafkaAdminClient adminClient, String name) {
        // 创建 Topic
        NewTopic newTopic1 = new NewTopic(
                name, 3, (short) 1
        );

        List<NewTopic> newTopics = Arrays.asList(newTopic1);

        // 创建 Topic
        adminClient.createTopics(newTopics);
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        createTopic();
    }
}
